package com.muchfish.reactor.api;

import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.UnicastProcessor;

public class ReactorCoreConceptsExample {
    public static void main(String[] args) {
        // 创建发布者
        Flux<Integer> source = Flux.range(1, 5);

        // 创建处理器，并进行数据处理
        UnicastProcessor<Integer> processor = UnicastProcessor.create();
        source.map(value -> value * 2)  // Example: doubling the values
                .subscribe(processor);

        // 创建订阅者
        CustomSubscriber<Integer> subscriber = new CustomSubscriber<>();

        // 订阅并处理数据
        processor.subscribe(subscriber);
    }

    // 自定义订阅者
    static class CustomSubscriber<T> extends BaseSubscriber<T> {
        @Override
        protected void hookOnNext(T value) {
            System.out.println("Processed Value: " + value);
        }

        @Override
        protected void hookOnError(Throwable throwable) {
            System.err.println("Error: " + throwable);
        }

        @Override
        protected void hookOnComplete() {
            System.out.println("Done");
        }
    }
}





